//!
//! FlightSQL Command structures
//!

use std::error::Error;
use std::fmt::Display;

use arrow_flight::sql::{ActionClosePreparedStatementRequest, ActionCreatePreparedStatementRequest, Any, CommandGetCatalogs, CommandGetCrossReference, CommandGetDbSchemas, CommandGetExportedKeys, CommandGetImportedKeys, CommandGetPrimaryKeys, CommandGetSqlInfo, CommandGetTableTypes, CommandGetTables, CommandGetXdbcTypeInfo, CommandPreparedStatementQuery, CommandStatementQuery, ProstMessageExt};
use arrow_flight::{FlightDescriptor, Ticket};
use bytes::Bytes;
use prost::Message;
use crate::error::{arrow_error_to_status, decode_error_to_status, FlightSQLError, Result};
use crate::util::*;

/// Decoded / validated FlightSQL command messages
///
/// Handles encoding/decoding prost::Any messages back and forth to native Rust types
///
#[derive(Debug, Clone, PartialEq)]
pub enum FlightSQLCommand {
    /// Run a normal query
    CommandStatementQuery(CommandStatementQuery),

    /// Get a list of the available tables
    CommandGetTables(CommandGetTables),
}

impl FlightSQLCommand {
    /// Figure out and decode the specific FlightSQL command in `msg` and decode it to a native IOx / Rust struct
    fn try_decode<B: bytes::Buf>(msg: B) -> Result<Self> {
        let any: prost_types::Any = prost::Message::decode(msg).map_err(decode_error_to_status)?;

        match any {
            _ if any.is::<CommandStatementQuery>() => {
                let command = any.unpack()?.expect("CommandStatementQuery try_decode error");
                Ok(FlightSQLCommand::CommandStatementQuery(command))
            }
            _ if any.is::<CommandGetTables>() => {
                let command = any.unpack().map_err(arrow_error_to_status)?
                    .expect("CommandGetTables try_decode error");
                Ok(FlightSQLCommand::CommandGetTables(command))
            }
            _ => Err(FlightSQLError::NotImplementedYet(format!(
                "still need to implement support for {}",
                any.type_url
            ))),
        }
    }

    pub fn try_parse_ticket(ticket: Ticket) -> Result<Self> {
        FlightSQLCommand::try_decode(&*ticket.ticket)
    }

    pub fn try_parse_flight_descriptor(
        flight_descriptor: FlightDescriptor,
    ) -> Result<Self> {
        FlightSQLCommand::try_decode(&*flight_descriptor.cmd)
    }

    /// Encode the command as a flightsql message (bytes)
    pub fn try_encode(self) -> Result<Bytes> {
        let msg = match self {
            Self::CommandStatementQuery(cmd) => Any::pack(&cmd),
            Self::CommandGetTables(cmd) => Any::pack(&cmd),
        }?;

        Ok(msg.encode_to_vec().into())
    }

    pub fn to_ticket(&self) -> Ticket {
        let ticket = match self {
            FlightSQLCommand::CommandStatementQuery(cmd) => cmd.as_any().encode_to_vec(),
            FlightSQLCommand::CommandGetTables(cmd) => cmd.as_any().encode_to_vec(),
        };
        Ticket {
            ticket: ticket.into()
        }
    }
}

#[cfg(test)]
mod test {
    #[test]
    fn test() {
        assert_eq!(1, 1);
    }
}